import org.apache.spark.sql.SparkSession

/*任务：
2.1、计算独立IP数

2.2、统计每个视频独立IP数（视频的标志：在日志文件的某些可以找到 *.mp4，代表一个视频文件）

2.3、统计一天中每个小时的流量
*/
object Q2_LogAnalysis {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .appName(this.getClass.getCanonicalName.init)
      .master("local[*]")
      .getOrCreate()
    val sc = spark.sparkContext
    sc.setLogLevel("ERROR")
    import spark.implicits._
    //读文件
    val lines = sc.textFile("F:\\TrainingCourse\\P1 大数据\\大数据正式班第四阶段模块二\\spark实战应用（上）\\代码\\data\\cdn.txt")
    val rawDF = lines.map(line => {
      val ip = line.split(" ")(0)
      val time = line.split("[\\[\\]]")(1)
      val addr = line.split("\"")(1).split(" ")(1)
      (ip, time, addr)
    }).toDF("IP", "TIME", "ADDRESS") //IP TIME ADDRESS
    rawDF.show(10)
    //1.1计算独立IP数
    val distinctIP = rawDF.select("IP").distinct().count()
    println("独立的IP数：" + distinctIP)

    //2.2、统计每个视频独立IP数（视频的标志：在日志文件的某些可以找到 *.mp4，代表一个视频文件）

    rawDF
      .filter(row => {
        row.getString(2).endsWith(".mp4")
      })
      .selectExpr("IP", "split(ADDRESS,'/') as ArrayAddr")
      .selectExpr("IP", "ArrayAddr[size(ArrayAddr)-1] as ADDRESS")
      .distinct()
      .groupBy("ADDRESS")
      .count()
      .sort('count.desc)
      .show(20)
/*      .coalesce(1)
      .write
      .mode("overwrite")
      .csv("result/csv")*/
    //3、统计一天中每个小时的流量
    rawDF.createOrReplaceTempView("r1")
    //直接滤除了不正常的时间格式
    spark.sql(
      """
        |select count(IP), hour(to_timestamp(TIME,"dd/MMM/yyyy:HH:mm:ss Z")) `HOUR`
        |from r1
        |group by HOUR
        |having HOUR >=0
        |order by HOUR
        |""".stripMargin).show(24)


/*    spark.sql(
      """
        |select *
        |from r1
        |where hour(to_timestamp(TIME,"dd/MMM/yyyy:HH:mm:ss Z")) < 0
        |""".stripMargin
    ).show(50)*/
    spark.stop()
  }
}
